package per.dg.rxjava.rx1;


import java.util.List;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func0;
import rx.functions.Func1;

/**
 * 创建Observable方法create、just、from、defer、interval、timer；
 * 对Observable传递的数据进行操作：map、flatMap、concatMap、buffer、filter
 */

public class RxPractice {

    public void original(){

    }

//    创建被观察者

    /**
     * 使用creat方法创建被观察者
     * 将数据发送给观察者
     * call方法会在观察者被订阅时触发
     * @param t 要向观察者传输的数据
     * @param <T> 向观察者传输的数据类型
     * @return 被观察者
     */
    private <T> Observable creatObservable_creat(final T t){
        Observable<T> observable = Observable.create(new Observable.OnSubscribe<T>() {
            @Override
            public void call(Subscriber<? super T> subscriber) {
//                向观察者发送数据
                subscriber.onNext(t);
//                通知观察者传输完毕，不会再传输数据
                subscriber.onCompleted();
            }
        });
        return observable;
    }

    /**
     * 使用just方法创建被观察者
     * 将数据依次发送给观察者
     * @return
     */
    private Observable creatObservable_just(){
        Observable observable = Observable.just("11", 22, 12.33);
        return observable;
    }

    /**
     * 与just创建的被观察者类似，执行just创建的方法与from创建执行的方法一样
     * just只是将不同的类型转成Object
     * @return
     */
    private Observable creatObservable_from(){
        String[] s = new String[]{"11", "22"};
        Observable<String> from = Observable.from(s);
        return from;
    }

    private void creatObservable_defer(){
        Observable.defer(new Func0<Observable<Integer>>() {
            @Override
            public Observable<Integer> call() {
                return Observable.create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onCompleted();
                    }
                });
            }
        });
    }

    /**
     * 通过interval方法创建被观察者
     * 定时调用onNext方法，向观察者发送定义的时长
     * @return
     */
    private Observable creatObservable_interval(){
        Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);
        return interval;
    }

    /**
     * 通过timer方法创建被观察者
     * 延迟一段时间（1000毫秒）后，向观察者发送数据
     * @return
     */
    private Observable creatObservable_timer(){
        Observable<Long> timer = Observable.timer(1000, TimeUnit.MILLISECONDS);
        return timer;
    }

//    创建被观察者


//    对被观察者发送的数据进行操作，再发送给观察者

    /**
     * 通过制定一个Func1对象,将原Observable对象转换为另一个Observable对象并发射。
     * @param observable 需要转换数据的被观察者
     * @return
     */
    private Observable changeData_map(Observable observable){
        Observable map = observable.map(new Func1() {
            @Override
            public Object call(Object o) {
                return 11;
            }
        });
        return map;
    }

    /**
     * 对被观察者发送的数据（集合，数组类）的每一项，进行处理，然后重新创建一个被观察者，依次把处理后的数据发送给观察者
     * 注意：并不一定会按照存储数据的顺序发送
     * @param observable
     * @return
     */
    private Observable changeData_flatmap(Observable<List<Integer>> observable){
        Observable observable1 = observable.flatMap(new Func1() {
            @Override
            public Object call(Object o) {
                return Observable.create(new Observable.OnSubscribe<Object>() {
                    @Override
                    public void call(Subscriber<? super Object> subscriber) {
                        subscriber.onNext(22);
                    }
                });
            }
        });
        return observable;
    }

    /**
     * 功能与flatmap一样，
     * 区别：发送的顺序与存储数据的顺序相同
     * @param observable
     * @return
     */
    private Observable changeData_concatMap(Observable<List<Integer>> observable){
        Observable observable1 = observable.concatMap(new Func1() {
            @Override
            public Object call(Object o) {
                return Observable.create(new Observable.OnSubscribe<Object>() {
                    @Override
                    public void call(Subscriber<? super Object> subscriber) {
                        subscriber.onNext(33);
                    }
                });
            }
        });
        return observable;
    }

    /**
     * 对依次发送的数据进行分组发送
     * @param observable
     * @return
     */
    private Observable changeData_buffer(Observable observable){
        Observable buffer = observable.buffer(3);
        return buffer;
    }

    /**
     * 按照指定的规则对数据进行筛选
     * 符合的情况返回true，不符合的返回false
     * @param observable
     * @return
     */
    private Observable changeData_fliter(Observable observable){
        Observable filter = observable.filter(new Func1() {
            @Override
            public Object call(Object o) {
                return true;
            }
        });
        return filter;
    }

//    对被观察者发送的数据进行操作，再发送给观察者

}
